相关配置文件

生产配置

#env.java.opts.jobmanager: -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=8085
#env.java.opts.taskmanager: -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=8086
rest.flamegraph.enabled: true
#jobmanage配置
jobmanager.archive.per-user: true
jobmanager.memory.process.size: 1600m
jobmanager.rpc.address: bigdata2

assloader.check-leaked-classloader: false
#taskmanage配置

taskmanager.memory.managed.fraction: 0.1
#taskmanager.memory.network.fraction: 0.1
taskmanager.memory.process.size: 3072m
#taskmanager.memory.jvm-overhead.max: 3072m
#taskmanager.memory.task.heap.size: 2048m
taskmanager.numberOfTaskSlots: 1
parallelism.default: 1
sink.shuffle-by-partition.enable: true
slotmanager.number-of-slots.max: 50


#yarn配置
yarn.application-attempts: 5
yarn.maximum-failed-containers: 100
yarn.tags: flink


#状态后端
state.backend: ROCKSDB
state.backend.incremental: true
state.backend.local-recovery: true
state.backend.rocksdb.memory.high-prio-pool-ratio: 0.1
state.backend.rocksdb.memory.managed: true
state.backend.rocksdb.memory.write-buffer-ratio: 0.5
state.backend.rocksdb.predefined-options: DEFAULT
state.backend.rocksdb.timer-service.factory: ROCKSDB
state.savepoints.dir: hdfs://bigdata/flink/savepoints

#检查点配置
execution.buffer-timeout: 100
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
state.checkpoints.num-retained: 10
execution.checkpointing.interval: 1min
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.min-pause: 1min
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.snapshot-compression: false
execution.checkpointing.timeout: 5min
execution.checkpointing.unaligned: false
execution.checkpointing.tolerable-failed-checkpoints: 10
state.checkpoints.dir: hdfs://bigdata/flink/checkpoints

#高可用,flink再1.15以后不再支持zk3.4要弄一个新的zk集群才能使用高版本的高可用
#high-availability: zookeeper
#high-availability.storageDir: hdfs://bigdata/flink/recovery
#high-availability.zookeeper.quorum: bigdata17:2181,bigdata18:2181,bigdata19:2181
#high-availability.zookeeper.path.root: /flink

#失败重启策略
jobmanager.execution.failover-strategy: region
restart-strategy: fixed-delay
restart-strategy.fixed-delay.attempts: 3
restart-strategy.fixed-delay.delay: 5s

#配置历史服务器
jobmanager.archive.fs.dir: hdfs://bigdata/flink_completed_jobs/
historyserver.web.address: bigdata2
historyserver.web.port: 8082
historyserver.web.ssl.enabled: false
historyserver.security.spnego.auth.enabled: false
historyserver.archive.fs.dir: hdfs://bigdata/flink_completed_jobs/
historyserver.cli.fallback: true
historyserver.archive.fs.refresh-interval: 10000

#jobmanager.scheduler: adaptive
#jobmanager.adaptive-scheduler.scaling-interval.min: 15s
#监控

#metrics.latency.interval: 60
#metrics.reporter.promgateway.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter

#metrics.reporter.promgateway.host: node2
#metrics.reporter.promgateway.port: 9091
#metrics.reporter.promgateway.jobName: wms
#metrics.reporter.promgateway.randomJobNameSuffix: true
#metrics.reporter.promgateway.deleteOnShutdown: false
#metrics.reporter.promgateway.interval: 20 SECONDS


env.java.opts.all: --add-exports=java.base/sun.net.util=ALL-UNNAMED --add-exports=java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-exports=java.security.jgss/sun.security.krb5=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.text=ALL-UNNAMED --add-opens=java.base/java.time=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED  -Dfile.encoding=UTF-8

classloader.check-leaked-classloader: false
classloader.resolve-order: child-first
#classloader.resolve-order: parent-first
classloader.parent-first-patterns.default: "org.apache.commons";"java.";"scala.";"org.apache.flink.";"com.esotericsoftware.kryo";"org.apache.hadoop.";"javax.annotation.";"org.xml";"javax.xml";"org.apache.xerces";"org.w3c";"org.rocksdb.";"org.slf4j";"org.apache.log4j";"ch.qos.logback"

创建对应的目录

hdfs dfs -mkdir -p /user/flink/savepoints
hdfs dfs -mkdir -p /user/flink/checkpoints
hdfs dfs -mkdir -p /user/flink/recovery
hdfs dfs -mkdir -p /user/flink/applicationHistory

配置文件注解参考

# 提交的方式是per-job
execution.target: yarn-per-job

# 状态过期时间
table.exec.state.ttl=10min

# JobManager 进程的最大可用内存为 2,147,483,648 字节，等于 2 GB。
jobmanager.memory.process.size: 2147483648

# 这个配置定义了缓冲时间（以毫秒为单位），用于控制Flink作业的数据缓冲。
execution.buffer-timeout: 100

# 这个配置控制检查点的外部保留策略，表示当作业被取消时，检查点数据将被保留，以便后续的分析或故障恢复。
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION

# 这个配置定义了检查点之间的时间间隔，以毫秒为单位。在此间隔内，Flink将尝试创建新的检查点以保存作业的状态。
execution.checkpointing.interval: 180000

表示检查点（checkpoint）的超时时间为600,000毫秒，或者等同于600秒。
execution.checkpointing.timeout: 600000

# 这个配置指定了允许同时运行的最大检查点数，通常设置为1以确保顺序检查点。
execution.checkpointing.max-concurrent-checkpoints: 1

# 这个配置定义了检查点的最小暂停时间，以毫秒为单位，通常用于确保不会频繁触发检查点。
execution.checkpointing.min-pause: 60000

# 这个配置指定了检查点模式，EXACTLY_ONCE 表示精确一次处理语义，用于确保数据不会重复或丢失。
execution.checkpointing.mode: EXACTLY_ONCE

# 这个配置用于控制检查点时是否进行快照压缩，通常关闭以减少性能开销。
execution.checkpointing.snapshot-compression: false

# 这个配置指定了作业执行的目标，"yarn-per-job" 表示为每个作业分配独立的YARN容器。
execution.target: yarn-per-job

# 这个配置指定了Flink的高可用性模式，这里使用Zookeeper。
high-availability: ZOOKEEPER

# 这个配置项定义了用于存储高可用性数据的目录，通常是HDFS路径。
high-availability.storageDir: hdfs:///user/flink/ha

# 这个配置项用于指定Zookeeper客户端访问控制列表（ACL）。
# 在这里，ACL被设置为 "open"，表示没有访问控制限制。
#high-availability.zookeeper.client.acl: open

# 这个配置项定义了用于存储Flink元数据的Zookeeper路径的根目录。
high-availability.zookeeper.path.root: /flink

# 这个配置项指定了Zookeeper的连接信息，包括Zookeeper服务器的主机和端口。
# 在这个例子中，Zookeeper集群包括三个服务器：10.251.37.43、10.251.37.44和10.251.37.45，它们监听在端口2181上。
high-availability.zookeeper.quorum: 10.251.37.43:2181,10.251.37.44:2181,10.251.37.45:2181

# 配置Flink的历史服务器（History Server）的归档文件存储目录，通常用于存储作业执行历史信息。
historyserver.archive.fs.dir: hdfs:///user/flink/applicationHistory

# 当历史服务器无法提供服务时，是否启用Flink CLI的回退模式。
historyserver.cli.fallback: true

# 配置JobManager的归档文件存储目录，用于存储有关Flink作业的元数据信息。
jobmanager.archive.fs.dir: hdfs:///user/flink/applicationHistory

# 如果启用，每个用户都有一个独立的作业归档文件存储目录。
jobmanager.archive.per-user: true

# 配置JobManager的JVM堆大小，这里设置为1 GB（1073741824字节）。
jobmanager.heap.size: 1073741824

# 是否启用Kerberos身份验证，这里设置为false，表示不启用Kerberos。
kerberos.auth.enabled: false

# 默认的并行度，用于作业中未指定并行度的操作符。
parallelism.default: 1

# 自动生成水印（watermark）的时间间隔，单位为毫秒。
pipeline.auto-watermark-interval: 200

# 启用通用类型检查，通常用于检测泛型类型错误。
pipeline.generic-types: true

# 是否启用对象重用，通常用于提高性能。
pipeline.object-reuse: false

# 指定用于 Python 客户端执行的可执行文件路径。
python.client.executable: /opt/cloudera/parcels/Anaconda/bin/python3

# 是否启用按分区的数据分发，通常用于提高性能。
sink.shuffle-by-partition.enable: true

# 配置最大可用的槽位数量，用于并行执行任务，这里设置为50。
slotmanager.number-of-slots.max: 50

# 配置Flink的状态后端（State Backend）为ROCKSDB，用于存储作业的状态信息。
state.backend: ROCKSDB

# 是否启用状态增量备份，通常用于减少检查点的开销。
state.backend.incremental: true

# 启用本地状态恢复，通常用于在故障时快速恢复状态。
state.backend.local-recovery: true

# ROCKSDB内存管理的高优先级内存池比例，通常用于提高性能。
state.backend.rocksdb.memory.high-prio-pool-ratio: 0.1

# 启用ROCKSDB内存管理，通常用于有效管理内存资源。
state.backend.rocksdb.memory.managed: true

# ROCKSDB写缓冲区的比例，通常用于控制写入性能。
state.backend.rocksdb.memory.write-buffer-ratio: 0.5

# 预定义的ROCKSDB选项，通常设置为DEFAULT以使用默认选项。
state.backend.rocksdb.predefined-options: DEFAULT

# ROCKSDB计时服务工厂，通常设置为ROCKSDB以使用ROCKSDB的计时服务。
state.backend.rocksdb.timer-service.factory: ROCKSDB

# 配置检查点（checkpoint）的目录，通常设置为HDFS路径，用于存储检查点数据。
state.checkpoints.dir: hdfs:///user/flink/checkpoints

# 保留的检查点（checkpoint）数目，通常设置为10，用于限制检查点的数量，旧的检查点会被自动删除。
state.checkpoints.num-retained: 10

# 配置保存点（savepoint）的目录，通常设置为HDFS路径，用于存储保存点数据。
state.savepoints.dir: hdfs:///user/flink/savepoints

# 任务管理器（TaskManager）的内存管理分数，通常设置为0.2，表示任务管理器分配给Flink内存的比例。
taskmanager.memory.managed.fraction: 0.2

# 任务管理器网络内存分数，通常设置为0.1，表示任务管理器分配给网络通信的内存比例。
taskmanager.memory.network.fraction: 0.1

# 任务管理器网络内存的最大大小，通常设置为2147483648字节（2 GB），用于限制网络内存的使用。
taskmanager.memory.network.max: 2147483648

# 任务管理器进程的大小，通常设置为2147483648字节（2 GB），用于限制任务管理器进程的内存使用。
taskmanager.memory.process.size: 2147483648

# 配置每个任务管理器（TaskManager）的任务槽（slots）数量，通常设置为1，表示每个任务管理器可以运行一个任务。
taskmanager.numberOfTaskSlots: 1

# YARN应用程序尝试重启的次数，通常设置为5，以确保在失败时有5次重试的机会。
yarn.application-attempts: 5

# YARN容器启动命令的模板，其中包括Java命令、JVM内存设置、JVM选项、日志和应用程序的参数。
yarn.container-start-command-template: %java% %jvmmem% %jvmopts% -Dlog4j2.formatMsgNoLookups=true -DyarnContainerId=$CONTAINER_ID %logging% %class% %args% %redirects%

# 允许的最大失败容器数，通常设置为100，以防止由于失败容器过多而导致资源耗尽。
yarn.maximum-failed-containers: 100

# 用于标记YARN应用程序的标签，通常设置为"flink"，用于标识Flink应用程序。
yarn.tags: flink

# 配置历史服务器（History Server）的Web地址，通常设置为主机名或IP地址。
historyserver.web.address: bj-yg-bi-prd-10-240-20-24-belle.lan

# 配置历史服务器的Web端口，通常设置为18211，用于访问历史服务器的Web界面。
historyserver.web.port: 18211

# 是否启用历史服务器的SSL安全连接，通常设置为false，表示不启用SSL。
historyserver.web.ssl.enabled: false

# 是否启用SPNEGO身份验证，通常设置为false，表示不启用SPNEGO身份验证。
historyserver.security.spnego.auth.enabled: false

flink写入clickhouse提速

akka.framesize：
804800000b
akka.ask.timeout：
50s

ytm：10240